package com.os.task;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.os.config.RsConfig;
import com.os.constant.Mq;
import com.os.model.LocationMsg;
import lombok.SneakyThrows;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;

/**
 * 定位数据多线程任务
 * @author 风亦未止
 * @date 2023/5/7 13:27
 */
@Log4j2
public class LocateData implements Callable<List<LocationMsg>> {

    private RabbitTemplate rabbitTemplate;
    private MessagePostProcessor messagePostProcessor;
    private RsConfig rsConfig;
    private String hash;

    public LocateData(RabbitTemplate rabbitTemplate, MessagePostProcessor messagePostProcessor, RsConfig rsConfig, String hash) {
        this.rabbitTemplate = rabbitTemplate;
        this.messagePostProcessor = messagePostProcessor;
        this.rsConfig = rsConfig;
        this.hash = hash;
    }

    @SneakyThrows
    @Override
    public List<LocationMsg> call() {
        rabbitTemplate.convertAndSend("data_exchange", null, hash, messagePostProcessor);
        List<LocationMsg> msgList = new ArrayList<>(rsConfig.TOTAL_SHARDS);
        for(int i = 1; i <= rsConfig.TOTAL_SHARDS; i ++){
            Message message = rabbitTemplate.receive(Mq.TEMP_QUEUE,500);
            if(ObjectUtil.isNull(message)){
                continue;
            }
            String msg = new String(message.getBody());
            LocationMsg locationMsg = JSONUtil.toBean(msg, LocationMsg.class);
            msgList.add(locationMsg);
        }
        return msgList;
    }

}
